package com.study.flink.source;

import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import com.study.flink.entity.PeterEntity;

import lombok.extern.slf4j.Slf4j;

/**
 * 自定义的数据源<br/>
 * 可以按照一定规律，随机产生数据。
 * 
 * @projectName: flink-demo
 * @author: peter
 * @description:
 * @date: 2019年8月7日 下午9:17:22
 * @version:
 */
@Slf4j
public class PeterSource implements SourceFunction<PeterEntity> {
    private static final long serialVersionUID = 1L;
    private static final String[] WORDS = new String[] {
        "Like", "many", "parts", "of", "the", "world", "Ethiopia", "has", "seen", "rapid",
        "deforestation", "over", "the", "past", "century", "Forest", "coverage", "in", "the",
        "East", "African", "nation", "has", "dropped", "from", "approximately", "35", "percent",
        "in", "the", "early", "1900s", "to", "just", "4", "percent", "in", "the", "2000s",
        "Ethiopians", "have", "made", "significant", "strides", "toward", "improving", "those",
        "numbers", "by", "planting", "350", "million", "trees", "in", "one", "day", "which",
        "Ethiopian", "officials", "believe", "is", "a", "new", "world", "record", "the", "BBC",
        "reports", "On", "July", "29", "volunteers", "from", "across", "the", "country", "got",
        "together", "with", "the", "goal", "of", "sowing", "200", "million", "seedlings", "before",
        "the", "day's", "end", "Certain", "public", "offices", "were", "closed", "Monday", "so",
        "government", "employees", "could", "participate", "Representatives", "from", "the",
        "United", "Nations", "the", "African", "Union", "and", "foreign", "embassies", "in",
        "Ethiopia", "took", "part", "in", "the", "project", "After", "12", "hours", "officials",
        "in", "charge", "of", "counting", "the", "seedlings", "confirmed", "that", "the",
        "planters", "had", "exceeded", "their", "initial", "goal:", "The", "country", "ended", "up",
        "planting", "350", "million", "trees", "on", "Monday", "breaking", "the", "previous",
        "record", "of", "66", "million", "set", "by", "15", "million", "volunteers", "in", "India",
        "in", "2017", "The", "initiative", "was", "part", "of", "Prime", "Minister", "Abiy",
        "Ahmed's", "Green", "Legacy", "campaign", "which", "aims", "to", "plant", "a", "total",
        "of", "4", "billion", "native", "trees", "in", "Ethiopia", "between", "May", "and",
        "October", "On", "a", "local", "scale", "the", "trees", "will", "improve", "soil",
        "quality", "and", "reduce", "the", "impact", "of", "flooding", "in", "the", "erosion-prone",
        "country", "The", "benefits", "of", "the", "new", "trees", "also", "extend", "beyond",
        "the", "nation's", "borders", "A", "single", "tree", "can", "absorb", "up", "to", "48",
        "pounds", "of", "carbon", "dioxide", "per", "year", "and", "sequester", "1", "ton", "of",
        "the", "greenhouse", "gas", "by", "the", "time", "it's", "40", "years", "old", "According",
        "to", "a", "recent", "study", "in", "Science", "by", "restoring", "90", "million", "square",
        "miles", "of", "the", "world's", "forests", "we", "could", "stop", "roughly", "226",
        "billion", "tons", "of", "carbon", "from", "becoming", "trapped", "in", "the", "Earth's",
        "atmosphere", "and", "warming", "the", "planet"
    };
    private static final int MAX_SIZE = WORDS.length;
    private static final int SLEEP_BASE_TIME = 500;
    private static final int SLEEP_TIME_RANGE = 100;
    private static final int OFFSET_RANGE = 3000;
    private static final int OFFSET_BASE = -1500;
    private static final int WATERMARK_INTERVAL = 10;

    private long time = 0;
    private Random rd = new Random();
    private long idx = 0;
    private ConcurrentLinkedQueue<Tuple3<Long, String, Long>> queue;
    
    public PeterSource() {
        queue = new ConcurrentLinkedQueue<Tuple3<Long, String,Long>>();
    }

    @Override
    public void run(SourceContext<PeterEntity> ctx) throws Exception {
        String word = "";
        long offset, time, sleepTime;

        while (true) {
            offset = OFFSET_BASE + rd.nextInt(OFFSET_RANGE);
            time = System.currentTimeMillis();
            word = WORDS[rd.nextInt(MAX_SIZE)];

            if (this.time > time + offset) {
                this.time = time + offset;
            }

            log.info(">>> {}, {}, {}", idx, word, time + offset);
            ctx.collectWithTimestamp(new PeterEntity(idx, word), time + offset);
//            queue.add(new Tuple3<Long, String, Long>(idx, word, time + offset));

            if (idx++ % WATERMARK_INTERVAL == 0) {
                ctx.emitWatermark(new Watermark(time));
            }
        }
    }

    @Override
    public void cancel() {
    }

}
